package com.test.study.ToolsUtils.strom.drpc;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
/**
 * 自定义bolt对请求的数据进行处理
 * @author huyong
 *
 */
@SuppressWarnings("serial")
public class MyBolt extends BaseRichBolt {
	@SuppressWarnings({ "rawtypes", "unused" })
	private Map stormConf;
	@SuppressWarnings("unused")
	private TopologyContext context;
	private OutputCollector collector;

	@SuppressWarnings("rawtypes")
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {

		this.stormConf=stormConf;
		this.context=context;
		this.collector=collector;
	}
/**
 * 这个input 分装了2个参数  第一个是参数的ID 第二个参数是参数的内容
 */
	@Override
	public void execute(Tuple input) {
		//获取到用户发送的数据
		String param = input.getString(1);//第一个参数是 参数0   第二个参数是1
		//对用户发送过来的数据进行处理
		StringBuffer stringBuffer = new StringBuffer();
		StringBuffer append = stringBuffer.append("hello_").append(param);
		//将数据发送出去到下一个bolt   这里的参数必须是2个  第一个参数就是前边传过来的参数ID  第二个参数就是 我们处理后的参数
		this.collector.emit(new Values(input.getLong(0),append.toString()));
		//发出去的数据 storm会自动使用return返回给调用者
		
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		//这里的字段是随便写的  ，他的接受方是通过角标获取的 ，但是不传是不行的
		declarer.declare(new Fields("id","value"));
		
		
	}
	



}
